package com.zjk.search.service;

import com.alibaba.fastjson.JSON;
import com.zjk.search.SearchItem;
import com.zjk.search.feign.SpuFeignClient;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: gfk
 * @create: 2022/10/26
 * @Description:
 * @FileName: EssyncService
 * @History:
 * @自定义内容：
 */
@Component
public class EssyncService {

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @Resource
    SpuFeignClient spuFeignClient;

    @Autowired
    RedisTemplate redisTemplate;

    @Value("${guli.elasticsearch.index}")
    private String index;

    @Scheduled(fixedRate = 5000)
    public void DB2ES() throws IOException {

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        System.out.println("Time=" + simpleDateFormat.format(new Date()));
        //将上次 同步的时间保存到Redis
         /*
        比如 每5分组同步一次
        同步时间                        16:30                  16:35                    16:40
        产品修改时间   16:26   16:29           16:31   16:32           16:36   16:37
         */
        String lastSyncTime = (String)redisTemplate.opsForValue().get("lastSyncTime");
        Map searchMap  = new HashMap<>();
        if(lastSyncTime!=null){
            searchMap.put("lastSyncTime",lastSyncTime);
        }
        List<SearchItem> searchItemList = spuFeignClient.findSearchItemList(searchMap);

        System.out.println("查询到增量数据：" + searchItemList.size());

        BulkRequest bulkRequest = new BulkRequest();//批处理
        searchItemList.stream().forEach(
                searchItem -> {
                    IndexRequest indexRequest = new IndexRequest(index);

                    indexRequest.id(searchItem.getSpuId().toString());//设置ID

                    String item = JSON.toJSONString(searchItem);
                    indexRequest.source(item, XContentType.JSON);//source

                    bulkRequest.add(indexRequest);
                }
        );

        if(bulkRequest.numberOfActions()>0){
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            System.out.println(bulkResponse.status());
        }

        //更新最后同步时间
        redisTemplate.opsForValue().set("lastSyncTime",simpleDateFormat.format(new Date()));
    }
}
